package com.klm.easymq.client.impl;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.klm.easymq.client.EasyMqClient;
import com.klm.easymq.config.EasyMqProperties;
import com.klm.easymq.core.IdempotentHandler;
import com.klm.easymq.core.MessageSerializer;
import com.klm.easymq.core.MonitorService;
import com.klm.easymq.entity.Message;
import com.klm.easymq.service.MessageService;
import lombok.extern.slf4j.Slf4j;

import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;

/**
 * Easy-MQ 客户端实现类
 * 
 * @author klm
 */
@Slf4j
public class EasyMqClientImpl implements EasyMqClient {
    
    private final EasyMqProperties properties;
    private final MessageService messageService;
    private final MessageSerializer messageSerializer;
    private final IdempotentHandler idempotentHandler;
    private final MonitorService monitorService;
    private final Executor asyncExecutor;
    
    public EasyMqClientImpl(EasyMqProperties properties,
                           MessageService messageService,
                           MessageSerializer messageSerializer,
                           IdempotentHandler idempotentHandler,
                           MonitorService monitorService,
                           Executor asyncExecutor) {
        this.properties = properties;
        this.messageService = messageService;
        this.messageSerializer = messageSerializer;
        this.idempotentHandler = idempotentHandler;
        this.monitorService = monitorService;
        this.asyncExecutor = asyncExecutor;
    }
    
    @Override
    public boolean sendMessage(String topic, Object message) {
        return sendMessage(topic, message, null);
    }
    
    @Override
    public boolean sendMessage(String topic, Object message, String businessKey) {
        try {
            String traceId = UUID.randomUUID().toString();
            byte[] messageBytes = messageSerializer.serialize(message);
            String messageContent = new String(messageBytes, java.nio.charset.StandardCharsets.UTF_8);
            
            // 创建消息记录
            Message messageEntity = new Message();
            messageEntity.setTraceId(traceId);
            messageEntity.setTopic(topic);
            messageEntity.setContent(messageContent);
            messageEntity.setBusinessKey(businessKey);
            messageEntity.setStatus("PENDING");
            messageEntity.setCreateTime(LocalDateTime.now());
            
            // 保存消息记录
            boolean saved = messageService.save(messageEntity);
            if (!saved) {
                log.error("Failed to save message: {}", traceId);
                return false;
            }
            
            // 记录监控 - 使用正确的方法名
            monitorService.recordMessageSend(traceId, topic, messageContent, "SENT");
            
            // 如果有业务键，进行幂等检查
            if (businessKey != null) {
                idempotentHandler.markProcessed(businessKey, (int) properties.getIdempotent().getCaffeine().getExpireAfterWrite());
            }
            
            log.info("Message sent successfully: topic={}, traceId={}, businessKey={}", topic, traceId, businessKey);
            return true;
            
        } catch (Exception e) {
            log.error("Failed to send message: topic={}, error={}", topic, e.getMessage(), e);
            return false;
        }
    }
    
    @Override
    public CompletableFuture<Boolean> sendMessageAsync(String topic, Object message) {
        return sendMessageAsync(topic, message, null);
    }
    
    @Override
    public CompletableFuture<Boolean> sendMessageAsync(String topic, Object message, String businessKey) {
        return CompletableFuture.supplyAsync(() -> sendMessage(topic, message, businessKey), asyncExecutor);
    }
    
    @Override
    public boolean sendBatchMessages(String topic, List<Object> messages) {
        try {
            for (Object message : messages) {
                if (!sendMessage(topic, message)) {
                    log.error("Failed to send batch message: topic={}", topic);
                    return false;
                }
            }
            return true;
        } catch (Exception e) {
            log.error("Failed to send batch messages: topic={}, error={}", topic, e.getMessage(), e);
            return false;
        }
    }
    
    @Override
    public Message getMessageByTraceId(String traceId) {
        try {
            return messageService.getById(traceId);
        } catch (Exception e) {
            log.error("Failed to get message by traceId: {}, error={}", traceId, e.getMessage(), e);
            return null;
        }
    }
    
    @Override
    public List<Message> getMessages(String topic, String status, int pageNum, int pageSize) {
        try {
            Page<Message> page = new Page<>(pageNum, pageSize);
            QueryWrapper<Message> queryWrapper = new QueryWrapper<>();
            
            if (topic != null && !topic.isEmpty()) {
                queryWrapper.like("topic", topic);
            }
            if (status != null && !status.isEmpty()) {
                queryWrapper.eq("status", status);
            }
            
            queryWrapper.orderByDesc("create_time");
            
            Page<Message> messagePage = messageService.page(page, queryWrapper);
            return messagePage.getRecords();
            
        } catch (Exception e) {
            log.error("Failed to get messages: topic={}, status={}, error={}", topic, status, e.getMessage(), e);
            return List.of();
        }
    }
    
    @Override
    public Map<String, Object> getMessageStats() {
        try {
            long totalCount = messageService.count();
            long successCount = messageService.count(new QueryWrapper<Message>().eq("status", "SUCCESS"));
            long errorCount = messageService.count(new QueryWrapper<Message>().eq("status", "ERROR"));
            
            return Map.of(
                "totalCount", totalCount,
                "successCount", successCount,
                "errorCount", errorCount,
                "successRate", totalCount > 0 ? (double) successCount / totalCount : 0
            );
            
        } catch (Exception e) {
            log.error("Failed to get message stats: error={}", e.getMessage(), e);
            return Map.of();
        }
    }
    
    @Override
    public boolean isMessageProcessed(String businessKey) {
        try {
            return idempotentHandler.isProcessed(businessKey);
        } catch (Exception e) {
            log.error("Failed to check message processed: businessKey={}, error={}", businessKey, e.getMessage(), e);
            return false;
        }
    }
    
    @Override
    public void markMessageAsProcessed(String businessKey, long expireSeconds) {
        try {
            idempotentHandler.markProcessed(businessKey, (int) expireSeconds);
        } catch (Exception e) {
            log.error("Failed to mark message as processed: businessKey={}, error={}", businessKey, e.getMessage(), e);
        }
    }
    
    @Override
    public IdempotentHandler getIdempotentHandler() {
        return idempotentHandler;
    }
    
    @Override
    public MonitorService getMonitorService() {
        return monitorService;
    }
    
    @Override
    public boolean isHealthy() {
        try {
            // 简单的健康检查：尝试查询一条消息
            messageService.count();
            return true;
        } catch (Exception e) {
            log.error("Health check failed: error={}", e.getMessage(), e);
            return false;
        }
    }
    
    @Override
    public void close() {
        try {
            // 关闭相关资源
            log.info("EasyMqClient is closing...");
        } catch (Exception e) {
            log.error("Failed to close EasyMqClient: error={}", e.getMessage(), e);
        }
    }
} 